package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.order.common.ChargingConstants;
import cn.tedu.charging.order.common.MqttConstants;
import cn.tedu.charging.order.common.OrderStatusConstants;
import cn.tedu.charging.order.dao.mapper.OrderSuccessMapper;
import cn.tedu.charging.order.dao.repository.ChargingProcessRepository;
import cn.tedu.charging.order.feign.CostClient;
import cn.tedu.charging.order.pojo.dto.ChargingResultDto;
import cn.tedu.charging.order.pojo.po.ChargingBillFailPO;
import cn.tedu.charging.order.pojo.po.ChargingBillSuccessPO;
import cn.tedu.charging.order.pojo.po.ChargingProcessPO;
import cn.tedu.charging.order.quartz.ApplicationContextProvider;
import cn.tedu.charging.order.websocket.ChargingWebSocketEndPoint;
import cn.tedu.common.pojo.JsonResult;
import cn.tedu.common.pojo.param.ChargingProcessParam;
import cn.tedu.common.pojo.utils.JsonUtils;
import cn.tedu.common.pojo.vo.ChargingProcessVO;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Queue;

/**
 *  EMQX的消息中间的 消费者
 *  订单服务 接收 设备发送 响应
 */
@Slf4j
public class MqttConsumer implements MqttCallbackExtended {

    //MqttConsumer 类中的 方法外面定义的变量 全局变量
    private MqttClient mqttClient;

    private MqttContext mqttContext;

    private OrderSuccessMapper abc;

    /**
     * 通过构造器 设置  容器里的 MqttClient
     * 赋值  MqttConsumer的 全局变量 mqttClient MqttConsumer 里的方法都能使用  mqttClient
     *  this.mqttClient = mqttClient;
     *
     *
     *
     * @param mqttClient
     */
    public MqttConsumer(MqttContext mqttContext, MqttClient mqttClient, OrderSuccessMapper orderSuccessMapper) {

        this.mqttClient = mqttClient;

        this.mqttContext = mqttContext;

        this.abc = orderSuccessMapper;
    }
    /**
     * 通过写方法 setter 设置 mqttClient
     * @param mqttClient
     */
    public void setMqttClient(MqttClient mqttClient) {
        this.mqttClient = mqttClient;
    }

    /**
     * 订单服务连接EMQX成功后，会调用这个方法
     * @param reconnect
     * @param serverURI
     */
    @SneakyThrows
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.debug("MqttConsumer#connectComplete方法被调用了,表示连接成功了,reconnect:{},serverURI:{}",
                reconnect,serverURI);
        //1.订阅启动充电的消息
        log.debug("连接成功,订阅 开始充电指令结果  topic:{}",MqttConstants.TOPIC_CHARGING_RESULT);
        //MqttConsumer 类中的 connectComplete 方法可以使用全局变量  mqttClient
        mqttClient.subscribe(MqttConstants.TOPIC_CHARGING_RESULT);
        log.debug("连接成功,成功 订阅开始充电指令结果 topic:{},",MqttConstants.TOPIC_CHARGING_RESULT);
        //2.订阅充电进度更新的消息
        log.debug("连接成功,用来接收设备同步的充电状态信息,订阅topic:{}",MqttConstants.TOPIC_CHARGING_PROCESS);
        mqttClient.subscribe(MqttConstants.TOPIC_CHARGING_PROCESS);
        log.debug("连接成功,用来接收设备同步的充电状态信息 订阅topic:{},成功.",MqttConstants.TOPIC_CHARGING_PROCESS);
    }

    @Override
    public void connectionLost(Throwable cause) {
        //  connectionLost 是无法使用  messageArrived 定义的局部变量  chargingResultDto
    }

    /**
     * 设备发送的消息 订单收到了 达到了 达到了订单服务
     * @param topic
     * @param message
     */
    @Override
    public void messageArrived(String topic, MqttMessage message)  {
        log.debug("MqttConsumer#messageArrived方法被调用了,表示消息到达,topic:{},message:{}",
                topic,message);
        if(MqttConstants.TOPIC_CHARGING_RESULT.equals(topic)) {
            handleChargingResult(message);//基于消息创建成功或失败订单
        }else if(MqttConstants.TOPIC_CHARGING_PROCESS.equals(topic)){
            handleChargingProcess(message);
        }
    }
    private void handleChargingProcess(MqttMessage message) {
        log.debug("处理设备发送的充电进度更新消息,message:{}",message);
        //1.基于openFeign接口远程调用计价服务
        //1.1构建参数对象
        ChargingProcessParam chargingProcessParam =
                JsonUtils.fromJson(message.toString(), ChargingProcessParam.class);
        log.debug("订单服务处理充电进度，构建参数对象:{}", chargingProcessParam);
        //1.2远程调用
        JsonResult<ChargingProcessVO> chargingProcessVOJsonResult =
                mqttContext.getCostClient().calculateCost(chargingProcessParam);
        log.debug("远程调用计价服务,返回结果:{}", chargingProcessVOJsonResult);
        //2.存储充电进度数据以及费用
        ChargingProcessVO chargingProcessVO = chargingProcessVOJsonResult.getData();
        ChargingProcessPO chargingProcessPO = new ChargingProcessPO();
        //数据拷贝
        BeanUtils.copyProperties(chargingProcessVO,chargingProcessPO);
        //时间
        chargingProcessPO.setId(System.currentTimeMillis());
        ChargingProcessRepository chargingProcessRepository =
                ApplicationContextProvider.getContext().getBean(ChargingProcessRepository.class);
        chargingProcessRepository.save(chargingProcessPO);
        log.debug("保存充电进度数据成功,chargingProcessPO:{}", chargingProcessPO);
        //3.推送充电数据给客户端(websocket)
        ChargingWebSocketEndPoint.sendAsyncMessage(chargingProcessParam.getUserId(),JsonUtils.toJson(chargingProcessPO));
    }

    private void handleChargingResult(MqttMessage message) {
        try {
            //把设备发送的消息JSON转换为java对象 chargingResultDto
            //chargingResultDto 定义在messageArrived 方法内部，局部变量，只能在方法里使用
            ChargingResultDto chargingResultDto =
                    JsonUtils.fromJson(message.toString(), ChargingResultDto.class);
            log.debug("接收到设备发送的开始充电指令结果,chargingResultDto:{}", chargingResultDto);
            if (chargingResultDto != null) {
                String result = chargingResultDto.getResult();
                log.debug("开始处理设备发送的开始充电指令结果:{}", result);
                if (ChargingConstants.RESULT_START_CHARGING_SUCCESS.equals(result)) {
                    log.debug("设备有响应,设备开始充电成功");
                    //保存订单记录
                    ChargingBillSuccessPO chargingBillSuccessPO = new ChargingBillSuccessPO();
                    chargingBillSuccessPO.setBillId(chargingResultDto.getOrderNo());
                    chargingBillSuccessPO.setOperatorId(chargingResultDto.getOperatorId());
                    chargingBillSuccessPO.setStationId(chargingResultDto.getStationId());
                    //Unknown column 'pile_id' in 'field list'
                    // 数据库表 charging_bill_success 没有pile_id ，保存的时候设置了 pile_id 数据库会报错
                    //chargingBillSuccessPO.setPileId(chargingResultDto.getPileId());
                    chargingBillSuccessPO.setGunId(chargingResultDto.getGunId());
                    chargingBillSuccessPO.setUserId(chargingResultDto.getUserId());
                    chargingBillSuccessPO.setVehicleId(chargingResultDto.getCarId());
                    //成功 订单状态为 充电中
                    chargingBillSuccessPO.setBillStatus(OrderStatusConstants.ORDER_STATUS_PROCESS);

                    //影响行数是指在执行SQL语句（如INSERT、UPDATE、DELETE）后，实际被操作的数据行的数量
                    //可以通过返回的影响行数来判断 这次数据库操作是否成功
                    // 先通过订单号 去数据库查询
                    log.debug("通过订单号:{} 去数据库查询", chargingResultDto.getOrderNo());

                    //QueryWrapper queryWrapper = new QueryWrapper<>();
                    //queryWrapper.eq("bill_id",chargingResultDto.getOrderNo());
                    //ChargingBillSuccessPO dbChargingSuccessPO =
                    //        mqttContext.getOrderSuccessMapper().selectOne(queryWrapper);
                    //if (dbChargingSuccessPO != null) {
                    //    log.debug("数据库中已经存在订单记录,billId:{}",chargingResultDto.getOrderNo());
                    //   return;
                    //}else  {
                    log.debug("数据库中不存在订单记录,billId:{},保存", chargingResultDto.getOrderNo());
                    log.debug("保存订单记录chargingBillSuccessPO:{}", chargingBillSuccessPO);
                    Integer row = mqttContext.getOrderSuccessMapper().insert(chargingBillSuccessPO);
                    log.debug("保存订单记录chargingBillSuccessPO:{},影响行数:{}", chargingBillSuccessPO, row);
                    // }
                    //abc.insert(chargingBillSuccessPO);
                } else {
                    log.debug("设备有响应,设备开始充电失败");
                    ChargingBillFailPO chargingBillFailPO = new ChargingBillFailPO();
                    chargingBillFailPO.setBillId(chargingResultDto.getOrderNo());
                    chargingBillFailPO.setOperatorId(chargingResultDto.getOperatorId());
                    chargingBillFailPO.setStationId(chargingResultDto.getStationId());
                    chargingBillFailPO.setVehicleId(chargingResultDto.getCarId());
                    //失败 需要记录失败原因
                    chargingBillFailPO.setFailDesc("设备有响应，开始充电失败");
                    log.debug("保存订单记录chargingBillFailPO:{}", chargingBillFailPO);
                    //todo 自己处理
                    int row = mqttContext.getOrderFailMapper().insert(chargingBillFailPO);
                    log.debug("保存订单记录chargingBillFailPO:{},影响行数:{}", chargingBillFailPO, row);
                    log.debug("通知充电的用户,表示歉意，提醒用户更换充电枪");
                    log.debug("通知充电的用户,表示歉意，给用户发送一张优惠券");
                    log.debug("通知运营商，提醒设备管理员，进行设备的检修");
                }
            }
        } catch (Exception e) {
            log.debug("处理设备发送的开始充电指令结果异常:{}", e);
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {

    }
}
